package com.we.risk.attribution.exposure.hive;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;

import static com.we.risk.attribution.AttributionInfo.*;

public class TableExpAttributionDataHive {
    public static final String LABELNAME = "label_name";
    public static final String LABELVALUE = "label_value";
    public static final String KFKTABLE = "adm.rp_exposure_attribution_data_sort_by_device";
    // 近30天数据
    public static final String QUERYSQL =
            "select device_id, "
                    + "unique_channel, "
                    + "cnt, "
                    + "receive_time, "
                    + "rt_type, "
                    + "hfq_advertiser_id, "
                    + "hfq_campaign_id, "
                    + "hfq_aid, "
                    + "hfq_click_id, "
                    + "UNIX_TIMESTAMP(dt, 'yyyy-MM-dd') as dt_time "
                    + "from "
                    + KFKTABLE
                    + " where dt >= '2023-02-10' "
                    + "and dt <= '2023-02-26'";

    public String device_id;
    public String unique_channel;
    public long cnt;
    public String receive_time;
    public int rt_type;
    public String hfq_advertiser_id;
    public String hfq_campaign_id;
    public String hfq_aid;
    public String hfq_click_id;
    public long dt_time;

    public TableExpAttributionDataHive() {}

    public TableExpAttributionDataHive(
            String device_id,
            String unique_channel,
            long cnt,
            String receive_time,
            int rt_type,
            String hfq_advertiser_id,
            String hfq_campaign_id,
            String hfq_aid,
            String hfq_click_id,
            long dt_time) {
        this.device_id = device_id;
        this.unique_channel = unique_channel;
        this.cnt = cnt;
        this.receive_time = receive_time;
        this.rt_type = rt_type;
        this.hfq_advertiser_id = hfq_advertiser_id;
        this.hfq_campaign_id = hfq_campaign_id;
        this.hfq_aid = hfq_aid;
        this.hfq_click_id = hfq_click_id;
        this.dt_time = dt_time;
    }

    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        tableEnv.useCatalog("cdh2_hive");
        TableConfig tableConf = tableEnv.getConfig();
        Configuration selfConf = new Configuration();
        selfConf.setBoolean("table.exec.hive.infer-source-parallelism", false);
        tableConf.addConfiguration(selfConf);
        Table tableRes = tableEnv.sqlQuery(TableExpAttributionDataHive.QUERYSQL);
        DataStream<TableExpAttributionDataHive> inputStream =
                tableEnv.toDataStream(tableRes, TableExpAttributionDataHive.class);

        SingleOutputStreamOperator<String> out =
                inputStream
                        .keyBy(
                                new KeySelector<TableExpAttributionDataHive, String>() {
                                    @Override
                                    public String getKey(TableExpAttributionDataHive value)
                                            throws Exception {
                                        return value.getDevice_id();
                                    }
                                })
                        .flatMap(
                                new RichFlatMapFunction<TableExpAttributionDataHive, String>() {
                                    @Override
                                    public void flatMap(
                                            TableExpAttributionDataHive value,
                                            Collector<String> out)
                                            throws Exception {
                                        String device_id = value.getDevice_id();
                                        String unique_channel = value.getUnique_channel();
                                        long dt_time = value.getDt_time();
                                        long recv_time = Long.parseLong(value.getReceive_time());
                                        long cnt = value.getCnt();

                                        int rt_type = value.getRt_type();
                                        String hfq_advertiser_id = value.getHfq_advertiser_id();
                                        String hfq_campaign_id = value.getHfq_campaign_id();
                                        String hfq_aid = value.getHfq_aid();
                                        String hfq_click_id = value.getHfq_click_id();

                                        JSONObject outJson = new JSONObject();
                                        outJson.put(DEVICEID, device_id);
                                        outJson.put(CHANNEL, unique_channel);
                                        outJson.put(DTTIME, dt_time);

                                        outJson.put(RTTYPE, rt_type);

                                        outJson.put(LABELNAME, RECVCNT);
                                        outJson.put(LABELVALUE, cnt);
                                        out.collect(outJson.toString());

                                        outJson.put(LABELNAME, RECVTIME);
                                        outJson.put(LABELVALUE, recv_time);
                                        out.collect(outJson.toString());

                                        outJson.put(LABELNAME, HFQADVSTID);
                                        outJson.put(LABELVALUE, hfq_advertiser_id);
                                        out.collect(outJson.toString());

                                        outJson.put(LABELNAME, HFQCMPGID);
                                        outJson.put(LABELVALUE, hfq_campaign_id);
                                        out.collect(outJson.toString());

                                        outJson.put(LABELNAME, HFQAID);
                                        outJson.put(LABELVALUE, hfq_aid);
                                        out.collect(outJson.toString());

                                        outJson.put(LABELNAME, HFQCLKID);
                                        outJson.put(LABELVALUE, hfq_click_id);
                                        out.collect(outJson.toString());
                                    }
                                });
        //                .flatMap(new browseOriginRichFlatMapFunc());

        return out;
    }

    public String getUnique_channel() {
        return unique_channel;
    }

    public long getDt_time() {
        return dt_time;
    }

    public String getDevice_id() {
        return device_id;
    }

    public long getCnt() {
        return cnt;
    }

    public String getReceive_time() {
        return receive_time;
    }

    public int getRt_type() {
        return rt_type;
    }

    public String getHfq_advertiser_id() {
        return hfq_advertiser_id;
    }

    public String getHfq_campaign_id() {
        return hfq_campaign_id;
    }

    public String getHfq_aid() {
        return hfq_aid;
    }

    public String getHfq_click_id() {
        return hfq_click_id;
    }
}
